import aiofiles
import asyncio
from typing import Union, List, Optional, Callable, AsyncIterable
from pathlib import Path
from aiopathlib import AsyncPath

from common.exception.errors import (
    ServerError,
    ServerNotFound
)
from common.log import log

class FileOperator:
    """文件操作类
    """
    
    @classmethod
    async def read_file_str(cls, file_path: str | Path) -> str:
        """以字符方式读取文件

        Args:
            file_path (str): 文件路径

        Returns:
            str: 文件内容
        """
        async with aiofiles.open(file_path, mode='r') as file:
            content = await file.read()
        return content
    
    @classmethod
    async def write_file_str(cls, file_path: str | Path, content: str) -> bool:
        """以字符方式写文件

        Args:
            file_path (str): 文件路径
            
        Returns:
            bool: True 文件写入成功
        """
        async with aiofiles.open(file_path, mode='w') as file:
            await file.write(content)
            
        return True
    
    @classmethod
    async def create_directory(cls, dir_path: str | Path) -> bool:
        """路径不存在则创建，否则不做处理 (非阻塞版本)"""
        path_obj = AsyncPath(dir_path)
        
        await path_obj.mkdir(parents=True, exist_ok=True)
        return True
        
    @classmethod
    async def read_file_binary(cls, file_path: str | Path) -> bytes:
        """以二进制方式读取文件

        Args:
            file_path (str): 文件路径

        Returns:
            bytes: 文件内容
        """
        async with aiofiles.open(file_path, mode='rb') as file:
            content = await file.read()
        return content
    
    @classmethod
    async def write_file_binary(cls, file_path: str | Path, content: bytes) -> bool:
        """以二进制的方式写文件

        Args:
            file_path (str): 文件路径
            content (bytes): 文件内容
            
        Returns:
            bool: True 文件写入成功
        """
        async with aiofiles.open(file_path, mode='wb') as file:
            await file.write(content)
        return True
            

    @classmethod
    async def list_directory(cls, dir_path: str | Path) -> list[str]:
        """列出目录下的文件（返回文件名列表）"""
        dir_path_obj = AsyncPath(dir_path)
        if await dir_path_obj.exists() and await dir_path_obj.is_dir():
            # 使用iterdir()遍历目录，并推导式获取文件名
            return [str(item) for item in dir_path_obj.iterdir() if await item.is_file()]
        else:
            raise NotADirectoryError(f"The path {dir_path} is not a valid directory.")
        
    @classmethod
    async def list_directory_recursively(
        cls, 
        dir_path: Union[str, Path], 
        list_files: bool = True, 
        list_dirs: bool = False
    ) -> List[str]:
        """递归列出目录下的所有文件/目录路径"""
        dir_path_obj = AsyncPath(dir_path)
        results = []
        if await dir_path_obj.exists() and await dir_path_obj.is_dir():
            for item in dir_path_obj.rglob('*'): # 使用 rglob 递归遍历
                if (list_files and await item.is_file()) or (list_dirs and await item.is_dir()):
                    results.append(str(item))
        return results
    
    @classmethod
    async def move_file(cls, src_path: Union[str, Path], dst_path: Union[str, Path]) -> None:
        src = AsyncPath(src_path)
        dst = AsyncPath(dst_path)
        await src.rename(dst)

    @classmethod
    async def copy_file(cls, src_path: Union[str, Path], dst_path: Union[str, Path]) -> None:
        # 对于大文件，可以考虑分块复制，类似你的 read_file_chunked 和写入组合
        async with aiofiles.open(src_path, 'rb') as src_file:
            content = await src_file.read()
        async with aiofiles.open(dst_path, 'wb') as dst_file:
            await dst_file.write(content)

    @classmethod
    async def rename_file(cls, file_path: Union[str, Path], new_name: str) -> None:
        path = AsyncPath(file_path)
        new_path = path.parent / new_name
        await path.rename(new_path)
    
    @classmethod
    async def exists(cls, path: str | Path) -> bool:
        """文件或路径是否存在"""
        return await AsyncPath(path).exists()
    
    @classmethod
    async def read_file_chunked(
        cls, 
        file_path: str | Path, 
        chunk_size: int = 1024 * 1024,  # 默认1MB
        progress_callback: Optional[Callable[[int, int], None]] = None
    ) -> AsyncIterable[bytes]:
        """
        流式分块读取大文件，支持进度回调
        
        Args:
            file_path: 文件路径
            chunk_size: 分块大小（字节）
            progress_callback: 进度回调函数，参数为(已读取字节数, 总字节数)
        
        Yields:
            bytes: 文件数据块
        """
        if not await cls.exists(file_path):
            raise FileNotFoundError(f"文件不存在: {file_path}")
        
        stat = await AsyncPath(file_path).stat()
        total_size = stat.st_size
        read_size = 0
        
        try:
            async with aiofiles.open(file_path, 'rb') as file:
                while True:
                    chunk = await file.read(chunk_size)
                    if not chunk:
                        break
                    
                    read_size += len(chunk)
                    if progress_callback:
                        progress_callback(read_size, total_size)
                    
                    yield chunk
        except Exception as e:
            log.error(f"分块读取文件 {file_path} 失败: {e}")
            raise ServerError(f"分块读取文件失败: {e}") from e
    
    @classmethod
    async def delete_file(
        cls, 
        files: Union[str, List[str], List[Path]], 
        suffixes: Optional[List[str]] = None,
        max_concurrent: int = 10,
    ) -> bool:
        """
        异步删除文件或多个文件，支持根据后缀名筛选
        
        Args:
            files: 文件路径或目录路径
            suffixes: 要删除的文件后缀名列表
            max_concurrent: 最大并发删除任务数
            cancellation_token: 取消令牌
        
        Returns:
            bool: 所有操作是否成功完成
        """
        # 统一处理输入参数
        if isinstance(files, str):
            paths = [files]
        else:
            paths = files
        
        semaphore = asyncio.Semaphore(max_concurrent)
        delete_tasks = []
        
        for path in paths:
            path_obj = AsyncPath(path)
                        
            # 检查路径是否存在
            if not await path_obj.exists():
                log.warning(f"路径不存在: {path}")
                continue
            
            # 如果是目录，处理目录中的所有文件
            if await path_obj.is_dir():
                for file_path in path_obj.iterdir():
                    if await file_path.is_file() and cls._delete_by_suffix(file_path.suffix, suffixes):
                        task = cls._delete_single_file_with_semaphore(file_path, semaphore)
                        delete_tasks.append(task)
            else:
                # 如果是单个文件，检查是否符合后缀要求
                if cls._delete_by_suffix(path_obj.suffix, suffixes):
                    task = cls._delete_single_file_with_semaphore(path_obj, semaphore)
                    delete_tasks.append(task)
        
        # 并行执行所有删除任务
        if delete_tasks:
            try:
                results = await asyncio.gather(*delete_tasks, return_exceptions=True)
                
                # 检查是否有异常发生
                for result in results:
                    if isinstance(result, Exception):
                        log.error(f"删除文件时发生错误: {result}")
                
                return True
            except asyncio.CancelledError:
                log.info("删除操作被取消")
                return False
        
        return True
    
    @classmethod
    def _delete_by_suffix(
        cls,
        file_suffix: str, 
        suffixes: list[str]
    ) -> bool:
        """检测后缀

        Args:
            file_suffix (str): 文件后缀
            suffixes (list[str]): 后缀列表

        Returns:
            bool: true 表示可以删除
        """    
        return not suffixes or file_suffix.lower() in [suffix.lower() for suffix in suffixes]
    
    @classmethod
    async def delete_in_directory(
        cls,
        directory: str | Path,
        suffixes: Optional[List[str]] = None,
        max_concurrent: int = 10,
    ):
        path = AsyncPath(directory)
        if not await path.exists() or not await path.is_dir():
            raise ServerNotFound(f"路径 {directory} 不存在")
        files = await cls.list_directory(directory)
                
        await cls.delete_file(files, suffixes, max_concurrent)

    @classmethod
    async def _delete_single_file_with_semaphore(
        cls, 
        file_path: Path | str | AsyncPath, 
        semaphore: asyncio.Semaphore,
    ) -> None:
        """
        带信号量控制的异步删除单个文件
        """
        async with semaphore:
            try:
                del_path = file_path
                if isinstance(file_path, str) or isinstance(file_path, Path):
                    del_path = AsyncPath(file_path)
                await del_path.unlink()
                log.info(f"已成功删除文件: {file_path}")
            except PermissionError as e:
                log.error(f"没有权限删除文件 {file_path}: {e}")
                raise
            except FileNotFoundError:
                log.warning(f"文件不存在，无需删除: {file_path}")
            except Exception as e:
                log.error(f"删除文件 {file_path} 时发生错误: {e}")
                raise